package com.feidee.fd.sml.algorithm.component.ml.ps

import com.feidee.fd.sml.algorithm.component.BasicParam
import com.feidee.fd.sml.algorithm.util.{Log, ToolClass}
import com.google.gson.Gson
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}


abstract class AbstractPSComponent[A <: BasicParam](implicit m: Manifest[A]) extends Log {
  val tool = new ToolClass()
  lazy val spark: SparkSession = SparkSession
    .builder.appName("psApp")
    .getOrCreate
  lazy val hiveSpark: SparkSession = SparkSession
    .builder
    .enableHiveSupport
    .getOrCreate

  /**
    * 参数解析
    * @param paramStr 加密参数
    * @return 解密后参数对象
    */
  def parseParam(paramStr: String): A = {
    require(tool.isNotNull(paramStr), "param can't be null")
    logInfo(s"encrypting parameter: $paramStr")
    val newParamStr = tool.decryption(paramStr)
    logInfo(s"decryption parameter: $newParamStr")
    new Gson().fromJson(newParamStr, m.runtimeClass)
  }

  /**
    * 数据加载
    * @return
    */
  def loadData(param: A): DataFrame = {
    val inputPaths = param.input_pt.split(",")
    var dataset = hiveSpark.read.load(inputPaths(0))
    for (i <- 1 until inputPaths.size) {
      dataset = dataset.union(hiveSpark.read.load(inputPaths(i)))
    }
    dataset
  }

  def process(param: A, data: DataFrame): DataFrame = data

  /**
    * 数据输出（覆盖）
    * @param data
    */
  def outputData(data: DataFrame, param: A): Unit = {
    data.write.mode(SaveMode.Overwrite).save(param.output_pt)
  }

  /**
    * 存结果到指定 Hive 表
    * @param data
    * @param param
    */
  def outputTable(data: DataFrame, param: A): Unit = {
    hiveSpark.sqlContext.sql(s"DROP TABLE IF EXISTS ${param.hive_table}")
//    data.sqlContext.sql(s"DROP TABLE IF EXISTS ${param.hive_table}")
    hiveSpark.createDataFrame(data.rdd,data.schema).write.saveAsTable(param.hive_table)
//    data.write.saveAsTable(param.hive_table)
  }

  def apply(paramStr: String): Unit = {
    val param = parseParam(paramStr)
    param.verify()
    val inputData = loadData(param)
    val outputDF = process(param, inputData)
    outputData(outputDF, param)
  }

}

